package com.runa.monitor.platinterface.mqtt;

import com.runa.monitor.comm.StormAIData.DownstreamMessage;
import com.runa.monitor.comm.dto.newComm.AsynResult;
import com.runa.persistence.SerializeUtil;
import com.runa.persistence.service.LogService;
import com.runa.persistence.util.LogUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.*;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 2024/7/24 16:23
 * zhangYB
 */
@Slf4j
public class MQTTAsyncClient extends MqttAsyncClient {

    private String clientId;
    private String type;
    private Integer index;

    private MqttAsyncClientPool pool;

    private BuildPublishConsumer callback;


    public MQTTAsyncClient(String serverURI, String clientId, String type, Integer index, MqttClientPersistence persistence, MQTTExecutor mqttExecutor, MqttAsyncClientPool mqttAsyncClientPool) throws MqttException {
        super(serverURI, String.format("%s-%s-%s", type, clientId, index), persistence);
        this.clientId = clientId;
        this.type = type;
        this.index = index;
        this.pool = mqttAsyncClientPool;
        this.callback =  new BuildPublishConsumer(mqttExecutor, this);
        setCallback(callback);
        connect(getOptions());
    }

    /**
     * 订阅消息
     *
     * @param topics
     * @throws MqttException
     */
    public void sub(List<String> topics) {
        if (!isConsumer()) {
            return;
        }
        if (!topics.isEmpty()) {
            String[] topicArr = topics.toArray(new String[0]);
            try {
                int[] qos = new int[topics.size()];
                Arrays.fill(qos, 1);
                subscribe(topicArr, qos);
            } catch (MqttException e) {

            }
        }
    }
    private AtomicInteger publishPendingCount =  new AtomicInteger(0);
    private AtomicInteger publishedCount =  new AtomicInteger(0);

    public void connectionLost() {
        publishPendingCount =  new AtomicInteger(0);
        publishedCount =  new AtomicInteger(0);
    }

    public void publishComplete(IMqttDeliveryToken token) {
//        decreasePublishPendingCount();
//        decreasePublishedCount();
    }

    private void decreasePublishPendingCount() {
        int i = publishPendingCount.get();
        while (!publishPendingCount.compareAndSet(i, i-1)) {
            i = publishPendingCount.get();
        }
    }
    private void decreasePublishedCount() {
        int i = publishedCount.get();
        while (!publishedCount.compareAndSet(i, i-1)) {
            i = publishedCount.get();
        }
    }
    private void increasePublishPendingCount() {
        int i = publishPendingCount.get();
        while (!publishPendingCount.compareAndSet(i, i+1)) {
            i = publishPendingCount.get();
        }
    }
    private void increasePublishedCount() {
        int i = publishedCount.get();
        while (!publishedCount.compareAndSet(i, i+1)) {
            i = publishedCount.get();
        }
    }

    public IMqttDeliveryToken publish(String topic, MqttMessage message)
            throws MqttException {
        IMqttDeliveryToken token = null;
        boolean result = true;
        try {
            token = this.publish(topic, message, null, null);
        } catch (MqttException e) {
            log.error("发送失败: {}", topic, e.getMessage());
            log.debug("发送失败: {}", e);
            result = false;
            MqttSchedule.addToQueue(topic, message);
        }
        if (result) {
            increasePublishedCount();
            if ("comm/net/onoffline".equals(topic)) {
                LogUtil.onofflineChange(message.getPayload());
            }
        } else {
            decreasePublishPendingCount();
        }
//        log.debug("{}-{}: 发送: {}, 已发: {}, ACK: {}, messageId: {} 飞行窗口(used): {}",
//                type, index, publishPendingCount.get(), publishedCount.get(), getPublishCompleteCount(),
//                token.getMessageId(), getInFlightMessageCount());
        return token;
    }

    private Boolean publishMessage(String topic, MqttMessage message) {
        try {
            if (!isConnected()) {
                MqttSchedule.addToQueue(topic, message);
                return false;
            }
            synchronized (this) {
                int count = publishPendingCount.get();
                int completeCount = getPublishCompleteCount();
                if (count - completeCount >= EmqConfig.getMaxInFlightCount()) {
                    MqttSchedule.addToQueue(topic, message);
//                    log.debug("发送数量: {}, 发送完成: {}, 飞行窗口: {}", count, completeCount, getInFlightMessageCount());
                    return false;
                }
                increasePublishPendingCount();
            }
            publish(topic, message);
            MqttSchedule.topicStatistics(topic);
        } catch (Exception e) {
            log.error("{}-{}: {} {} {} MQTT publishMessage失败 TOPIC: {} err: {}", type, index, publishPendingCount.get(), publishedCount.get(), getInFlightMessageCount(), topic, e.getMessage());
            log.debug("MQTT publishMessage失败: ", e);
            return false;
        }
        return true;
    }


    private Boolean publishBytes(String topic, byte[] bytes) {
        MqttMessage mqttMessage = new MqttMessage();
        if (bytes.length == 0) {
            log.error("消息体不能为空, topic:{}", topic);
            return false;
        }
        mqttMessage.setQos(1);
        mqttMessage.setPayload(bytes);
        return publishMessage(topic, mqttMessage);
    }


    public Boolean sendBytes(String topic, byte[] bytes) {
        try {
            return publishBytes(topic, bytes);
        } catch (Exception e) {
            log.error("bytes上报mqtt报错：{} {} ", topic, SerializeUtil.bytes2Obj(bytes, String.class), e);
        }
        return false;
    }

    public void sendAsyncResult(String topic, AsynResult asynResult) {
        boolean result;
        try {
            result = publishBytes(topic, Objects.requireNonNull(SerializeUtil.obj2Bytes(asynResult)));
        } catch (Exception e) {
            log.error("asynResult上报mqtt报错：{} {} ", topic, SerializeUtil.obj2json(asynResult), e);
            result = false;
        }
        LogService.upCount(asynResult.getTask(), asynResult.getId(), result);
    }


    public void sendMessage(String topic, MqttMessage message) {
        publishMessage(topic, message);
    }

    public void sendMessage(String topic, DownstreamMessage message) {
        String json = SerializeUtil.obj2json(message);
        if (StringUtils.isEmpty(json)) {
            log.error("消息体不能为空");
            return;
        }
        try {
            publishBytes(topic, json.getBytes());
            log.debug("TOPIC:{}------message:{}", topic, json);
        } catch (Exception e) {
            log.error("MqttException MQTT sendMessage信息有误---TOPIC:{}, message:{}, error: {}", topic, json, e.getMessage());
            log.debug("", e);
        }
    }

    private static MqttConnectOptions getOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);
        options.setUserName(EmqConfig.getUserName());
        options.setPassword(EmqConfig.getPassWord().toCharArray());
        options.setMaxInflight(EmqConfig.getMaxInFlightCount());
//        options.setMaxInflight(1);
        options.setServerURIs(new String[]{EmqConfig.getHOST()});
//        options.setServerURIs(new String[]{"tcp://192.168.1.102:1883"});
        // 设置超时时间
        options.setConnectionTimeout(EmqConfig.getTimeout());
        // 设置会话心跳时间
        options.setKeepAliveInterval(EmqConfig.getKeepalive());
        //设置断开后重新连接
        options.setAutomaticReconnect(true);
        return options;
    }


    public int getPublishCompleteCount() {
        return callback.getPublishCompleteCount();
    }

    private boolean isConsumer() {
        return "C".equalsIgnoreCase(type);
    }

}
